package gbench.sandbox.record.protein;

import org.junit.jupiter.api.Test;

import gbench.commonApp.data.DataApp;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.SQL;

import static gbench.common.tree.LittleTree.MFT;
import static gbench.common.tree.LittleTree.IRecord.*;
import static gbench.common.tree.LittleTree.SQL.nsql;
import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.Tuple2.*;
import static gbench.common.tree.LittleTree.Jdbc.IJdbcSession;

import java.time.LocalDateTime;
import java.util.DoubleSummaryStatistics;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 
 * @author gbench
 *
 */
public class JunitMapReduce {
    
    // 数据操作类
    final class MyDataApp extends DataApp {
        public MyDataApp(String dbname) { this.reload(this.configureDbName(dbname));}
        public IRecord configureDbName(final String dbname) { return 
            cfg.derive("url", MFT("jdbc:mysql://localhost:3306/{0}?serverTimezone=GMT%2B8", dbname));
        }
    }; // MyDataApp 数据数据库操作类
    
    /**
     * 求和
     * 
     * @param stream
     * @param mapper
     * @return
     */
    public Double sum(final Stream<?> stream, final Function<IRecord, Double> mapper) {
        
        if(stream==null)return 0d;
        return stream.map(e -> (IRecord) e).filter(e->e!=null).collect(Collectors.summarizingDouble(e -> {
            final var d = mapper.apply(e);
            return d == null? 0d:d;
        })).getSum();
    }
    
    /**
     * 求和
     * @param stream
     * @param mapper
     * @return
     */
    public DoubleSummaryStatistics dblstats(Stream<?> stream, Function<IRecord, Double> mapper) {
        return stream.map(e -> (IRecord) e).collect(Collectors.summarizingDouble(e -> mapper.apply(e)));
    }

   /**
    * 分解科目名称结构
    * @param line
    * @return 分解科目名称结构
    */
   final Tuple2<Integer,String> parseName(final String line) {
       var n = 0;
       var s = line;
       
       switch(line.charAt(0)) {
           case '-':{// 贷方
               n=1;
               s = s.substring(1);
               break;}
           case '?':{ // 未知
               n=2;
               s = s.substring(1);
               break;}
           default :{ // 借方
               break;}
       }//
       
       return TUP2(n,s);
   }
   
   /**
    * policy2entries
    * @param policy 记账策略
    * @return 复式记账法
    */
   public Stream<IRecord> policy2entries(List<IRecord> policy){
       final var epsilon = 1e-15;
       final var transcode = UUID.randomUUID();// 交易代码
       final var rr = policy.stream().map( p -> p.derive ( 
           REC(  "transcode", transcode, 
                 "amount", p.i4("drcr") == 2 ? 0d : RNDINT(1000), 
                 "createtime", LocalDateTime.now() )
       ));
       final var dd = rr.collect(Collectors.groupingBy(e -> e.str("drcr")));
       final var mm = REC(dd).aov2rec( (List<IRecord> vv) -> sum( vv.stream(), e->e.dbl("amount")) );
       
       if(mm.has("2")) {// 调节 2 
           final var d = mm.dbl("0")-mm.dbl("1"); // 提取
           final var aa = dd.get("2");
           aa.stream().forEach(r->{
               r.set("amount", Math.abs(d));
               r.set("drcr", d > 0 ? 1 : 0);
           });// dd
       }//if
       
       dd.forEach((k,vv)->{
           System.out.println(vv);
       });
       
       System.out.println();
       return dd.values().stream().flatMap(List::stream).filter(e -> e.dbl("amount") > epsilon);
   }
    
    /**
     * 求和
     */
    @Test
    public void qux() {
        final var dataApp = new MyDataApp("db1");
        final var proto = REC("#id",1,"transcode","","name","","drcr",0,"amount",1.0,"createtime",LocalDateTime.now());
        final var sb = SQL.builder("entry");// 狗仔entry sql 系列构建器具
        
        // 记账法
        final var accountingsPolicies = S("银行存款,-营业收入,?应交税金","库存商品,-营业成本,?成本调整")
            .map(e->e.split(",")) // 提取科目项目
            .map(ee->S(ee).map(e->{ final var t = parseName(e);
                return REC(
                    "drcr",t._1(),
                    "name",t._2() 
                );}).collect(Collectors.toList())
            ).collect(Collectors.toList());
        
        accountingsPolicies.forEach(ss->{
            ss.forEach(System.out::println);
            System.out.println();
        });
        
        final var createsqls = SQL.of("entry",proto).createTable(true);
        dataApp.withTransaction(sess->{
            //if(!sess.sql2get(MFT("show tables like ''{0}''","entry"),Boolean.class)) 
            {
                TUPLES(createsqls).filter(t->t._1()!=1).map(t->t._2()).forEach(trycatch2(sess::sqlexecute));
            }
            
            final var sqls = LIST(NATS(2000).map(e -> rnd(accountingsPolicies)).flatMap(this::policy2entries).map(sb::get));
            if (System.currentTimeMillis() < 0) {
                sqls.stream().map(SQL::insert).forEach(trycatch2(sess::sqlexecute));
            } else {
                final var recs = LIST(sqls.stream().map(e -> e.getSqlCtxAsOneRecord()));// 提取数据
                final var keys = recs.get(0).keys();
                final var s1 = JOIN(keys, ",");
                final var s2 = JOIN(keys.stream().map(e -> "e." + e), ",");
                final var line = "insert into entry (" + s1 + ") values ${ foreach e in entries (" + s2 + ") }";
                final var sql = nsql(line, REC("entries", recs));
                sess.sqlexecute(sql);
            } // if
            
            sess.sql2records("select * from entry").stream() // 提取结果
            .collect(pvtclc("name,drcr"))
            .dfs_eval_forall((path,ones)->{
                final var sts = dblstats(ones,e->e.dbl("amount"));
                return REC("path",path,"amount",sts.getSum(),"cnt",sts.getCount());
            }).forEach(System.out::println);
            
        });
    }
    
    @Data @AllArgsConstructor @NoArgsConstructor
    public static class DataEntry{
        String year;
        String month;
        Double total;
        public String toString() {
            return MFT("{0}-{1}/{2,number,#}",year,month,total) ;
        }
    }
    
    @Data @AllArgsConstructor @NoArgsConstructor
    public static class StatsItem{
        Double avg;
        Double count;
        Double sum;
        public String toString() {
            return MFT("avg:{0},count:{1},total:{2,number,#}",avg,count,sum) ;
        }
    }
    
    /**
     * 数据透视
     * @param year 数据年份
     * @param month 数据月份
     * @return
     */
    public List<IRecord> pivotTable(int year,int month) {
        final var dataApp = new MyDataApp("salepro");
        final var ar  = new AtomicReference<List<IRecord>>();
        final var proto = REC("year",year,"month",month);
        
        dataApp.withTransaction(sess->{
            final var sql = nsql("select goods_name,clientele_name,total from `order` "
                + "where year(create_at)=#year and month(create_at)=#month",
                REC("year",year,"month",month));
            final var recs = sess.sql2records(sql);
            final var rec = recs.stream()
                .collect(pvtclc("goods_name"))
                .dfs_eval_forall((path,vv)->REC(
                    MFT("{0}",path,year),
                    proto.derive("total",sum(vv,v->v.dbl("total"))).mutate(DataEntry.class)
                ));// dfs_eval_forall
            ar.set(rec);
        });// withTransaction
        
        return ar.get();
    }
    
    /**
     * mapreduce
     */
    @Test
    public void mapreduce() {
        final var months = A(1,2,3,4,5,6,7,8,9,10,11,12);
        final var recs = Stream.of(2017,2018,2019)
            .flatMap(e->cph(A(e),months).stream().map(p->TUP2(p.i4("0"),p.i4("1"))))
            .parallel()
            .flatMap(e->pivotTable(e._1(),e._2()).stream());
        final var r2019 = REC3(recs);
        r2019.dfs_eval_forall((p,v)->{
            return REC("path",p,"v",LIST(v));
        }).forEach(r->{
            System.out.println(r.get("path")+"--->"+r.lla("v",DataEntry.class));
        });
    }
    
    @Test
    public void db2() {
        // create database db2
        final var TBL_NUM = 100; // 数据表个数
        final var ROW_NUM = 100000; // 最大数据行数
        final var dbname = "db2"; // 数据库名
        final var dataApp = new MyDataApp(dbname);
        final var proto = REC("#id",1,"name","甲苯","amount",2d,"createtime",LocalDateTime.now());
        final var names = "甲苯,甲醇,乙醇,丁二醇".split(",");
        final var tables = LIST(NATS(TBL_NUM).map(i -> MFT( "T_ORDER_{0}", i)));
        
        final Consumer<Tuple2<IJdbcSession<UUID, Object>,String>> data_prepare = tup -> { // 表数据插入
            final var sess = tup._1(); // 会话
            final var tblname = tup._2(); // 表名
            final var last = timeit( () -> { // 计时统计
                final var sb = SQL.builder(tblname);// 狗仔entry sql 系列构建器具
                final var rb = IRecord.builder(proto.filter2("#id"));// 去除id
                final var createsqls = sb.get(proto).createTable(true); // 生成createsqls
                
                // 表创建语句
                TUPLES(createsqls).filter(t->t._1()!=1).map(t->t._2()).forEach(trycatch2(sess::sqlexecute));
                
                // 数据插入
                if(System.currentTimeMillis()<0) {
                    // 数据插入
                    NATS(ROW_NUM).map(i -> A(rnd(names), (int)(Math.random() * 1000), LocalDateTime.now()))
                    .map(aa -> sb.get(rb.get(aa))).forEach(trycatch2(sess::sqlexecute));
                }else {
                    // 数据插入
                    final var recs = LIST(NATS(ROW_NUM).map(i -> A(rnd(names), (int)(Math.random() * 1000), LocalDateTime.now())).map(rb::get));
                    final var keys = recs.get(0).keys(); // 获得keys
                    final var s1 = JOIN(keys, ",");
                    final var s2 = JOIN(keys.stream().map(e -> "e." + e), ",");
                    final var line = "insert delayed into "+tblname+" (" + s1 + ") values ${ foreach e in entries (" + s2 + ") }";
                    Stream.of(nsql(line, REC("entries", recs))).forEach(trycatch2(sess::sqlexecute));
                }//if
                
                // 增加name
                // sess.sql2stream(MFT("select * from {0}",tblname)).collect(pvtclc("name")).dfs_forall2(System.out::println);
            });// timeit
            
            // 显示数据执行时间
            System.out.println(MFT("{0} last for {1} s\n",tblname,last/1000));
        }; // data_prepare
        
        final var prepare_last = timeit( () -> {
            if( System.currentTimeMillis() < 0 )
                dataApp.withTransaction( sess -> tables.parallelStream().forEach(tbl->data_prepare.accept(TUP2(sess,tbl))) );
            else
                tables.parallelStream().map(tbl -> TUP2( tbl,new MyDataApp(dbname) ))
                .forEach( tup -> tup._2().withTransaction(sess->data_prepare.accept(TUP2(sess,tup._1()))) );
        }); // withTransaction
        
        System.out.println(MFT("数据准备 持续 : {0} s\n",prepare_last/1000));
        
        // 多表数据查询
        final var stats_last = timeit(()->{
            dataApp.withTransaction( sess -> {
                tables
                .stream()
                .map( table -> MFT("select ''{0}'' tblname,tbl.* from {0} tbl", table) ) // 多表查询
                .map( trycatch(sess::sql2stream) )
                .map( rr -> rr.collect(pvtclc("name")) )
                .flatMap( pvttbl -> {
                    final var pathItem = REC();
                    return pvttbl.dfs_eval_forall((path,vv) -> {
                        final var dss = dblstats(vv,v->v.dbl("amount")); // 数据统计
                        final var data = pathItem
                            .derive( "avg",dss.getAverage(),"count",dss.getCount(),"sum",dss.getSum() )
                            .mutate( StatsItem.class );
                        return REC( MFT("{0}",path), data );
                    }).stream();
                }) // flatMap
                .collect( llclc(IRecord::REC3) )
                .dfs_forall2( ( Tuple2<String,List<StatsItem>> tup ) -> {
                    final var path = tup._1();
                    final var sum = tup._2().stream().collect(Collectors.summarizingDouble(e->e.getSum())).getSum();
                    final var count = tup._2().stream().collect(Collectors.summarizingDouble(e->e.getCount())).getSum();
                    System.out.println(path+" ----> sum:"+sum+", count:"+count);
                }); // dfs_forall2
            }); // withTransaction
        });// 数值执行
        
        System.out.println(MFT("数据统计 持续 : {0} s\n",stats_last/1000));
        System.out.println(MFT("\n总体程序持续时间 : {0} s\n",(prepare_last+stats_last)/1000));
    }
    
    @Test
    public void foo() {
        final var TBL_NUM = 100;
        final var tables = LIST(NATS(TBL_NUM).map(i -> MFT( "T_ORDER_{0}", i)));
        final var s = tables.stream()
            .map(tbl->MFT("select name,count(*) cnt from {0} group by name",tbl))
            //.map(tbl->MFT("DROP TABLE {0}",tbl))
            //.collect(Collectors.joining(";\n "));
            .collect(Collectors.joining("\nunion "));
        System.out.println(s);
    }
    
}
